package com.coffee.test.util;

import com.coffee.netty.util.RemotingUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class NotifyProxy {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(NotifyProxy.class);

    public static NotifyProxy notifyProxy = new NotifyProxy();

    private AtomicInteger atomicInteger = new AtomicInteger(0);
    private ConcurrentHashMap<String, SyncZebraFuture> hashMap = new ConcurrentHashMap<>(4);


    //同步 获取返回的值 否则没有对应值； 【任何地方 调用 获取同步消息；那么就必须再Handler 里面 去 调用 ackSyncMsg 把返回消息放入SyncZebraFuture 获取】
    public String syncMsg(Channel cxt, String command) {
        logger.info("Send Sync message: {} {}", command, cxt.toString());
        SyncZebraFuture syncFuture = new SyncZebraFuture();
//        int key = atomicInteger.incrementAndGet();
//        command.addRequestId(key);
        hashMap.put(command, syncFuture);
        pushMessage(cxt, command +"\r\n");

        String result = null; // waite
        try {
            result = syncFuture.get(8, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            logger.error("Send syncMsg error ", e);
        }
        return result;
    }

    //*** 这个需要在handler 里面去 获取同步消息 放入SyncZebraFuture
    public void ackSyncMsg(Channel channel, String command) {
        logger.info("ACK Sync message: {} {}", command, channel.toString());
        SyncZebraFuture syncFuture = hashMap.get(command);   // 从缓存中获取数据
        if(syncFuture != null) {
            syncFuture.setResponse(command);
            hashMap.remove(command);   //主动释放
        }else {
            logger.error("SyncZebraFuture is null {}", channel.toString());
        }
    }

    public void pushMessage(Channel cxt, String message) {
        ChannelFuture cf = cxt.writeAndFlush(message);
        cf.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    if (logger.isDebugEnabled()){
                        logger.debug("push msg success: {} {}", message.toString(), RemotingUtil.channelToString(cxt));
                    }
                } else {
                    logger.error("push Message failure: {} {}", message.toString(), RemotingUtil.channelToString(cxt));
                }
            }
        });
    }



}
